package cn.hesion.log;
/**
 * ClassName: LogCollectorTask <br/>
 * Description: <br/>
 * date: 2021/1/27 18:12<br/>
 *
 * @author Hesion<br />
 * @version
 * @since JDK 1.8
 */

import cn.hesion.common.Constant;
import cn.hesion.singlton.PropTool2;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

import java.io.File;
import java.io.FileFilter;
import java.io.FilenameFilter;
import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Properties;
import java.util.TimerTask;

/**
 * @program: ClientDemo
 * @description:
 * @author: hesion
 * @create: 2021-01-27 18:12
 **/
public class LogCollectorTask extends TimerTask {



    @Override
    public void run() {
        Properties prop = null;
        try {
            prop = PropTool2.getProp();
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 采集日志的逻辑
        // 将文件先转移到一个临时目录里
        // 便利特定文件夹下的文件
        File logsFiles = new File(prop.getProperty(Constant.LOGS_DIR));
        String log_prefix = prop.getProperty(Constant.LOG_PREFIX);
        File[] uploadsFiles = logsFiles.listFiles(new FilenameFilter() {
            @Override
            public boolean accept(File dir, String name) {
                return name.startsWith(log_prefix);
            }
        });

        // 转移到临时的目录
        File tmpFile = new File(prop.getProperty(Constant.LOG_TMP_FOLDER));
        // 判断该文件夹是否存在，存在则继续，不存在则创建文件夹
        if(!tmpFile.exists()){
            tmpFile.mkdirs();
        }

        //转移日志到临时文件
        for (File file : uploadsFiles) {
            file.renameTo(new File(tmpFile.getPath()+"/"+file.getName()));
        }

        //上传到HDFS
        //1获取文件系统
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://linux121:9000");

        FileSystem fs = null;
        // 日期分门别类的存放
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        String todayStr = sdf.format(new Date());
        //配置在集群上运行
        try {
            fs = FileSystem.get(conf);
            //判断hdfs目标路径是否存在，备份目录是否存在
            Path path = new Path(prop.getProperty(Constant.HDFS_TARGET_FOLDER) + todayStr);
            // 判断hdfs目录是否存在,备份路径是否存在
            if(!fs.exists(path)){
                fs.mkdirs(path);
            }
            File bkFolder = new File(prop.get(Constant.BAK_FOLDER)  + todayStr);
            if(!bkFolder.exists()){
                bkFolder.mkdirs();
            }
            File[] files = tmpFile.listFiles();
            for (File file : files) {
                //上传到dfs
                String hdfsPath = prop.getProperty(Constant.HDFS_TARGET_FOLDER) + todayStr + "/" + file.getName();
                fs.copyFromLocalFile(new Path(file.getPath()),new Path(hdfsPath));
                file.renameTo(new File(bkFolder.getPath()+"/"+file.getName()));
            }

        } catch (Exception e) {
            e.printStackTrace();
        }




    }
}
